package cn.signalpha.modules.queue.utils;

import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.signalpha.core.tool.utils.StringUtil;
import lombok.AllArgsConstructor;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@AllArgsConstructor
public class RedisDelayQueueUtil {

	private final RedissonClient redissonClient;

	/**
	 * 添加延迟队列
	 * @param value 队列值
	 * @param endTime 结束时间
	 * @param queueCode 队列键
	 * @param <T> 泛型
	 */
	public <T> void addDelayQueue(T value, Date endTime, String queueCode) {
		long seconds = DateUtil.between(DateUtil.date(), endTime, DateUnit.SECOND);
		if (seconds > 0) {
			addDelayQueue(value, seconds, TimeUnit.SECONDS, queueCode);
		}
	}

	/**
	 * 添加延迟队列
	 * @param value 队列值
	 * @param endTime 结束时间
	 * @param queueCode 队列键
	 * @param <T> 泛型
	 */
	public <T> void addDelayQueue(T value, LocalDateTime endTime, String queueCode) {
		long seconds = Duration.between(LocalDateTime.now(), endTime).getSeconds();
		if (seconds > 0) {
			addDelayQueue(value, seconds, TimeUnit.SECONDS, queueCode);
		}
	}

	/**
	 * 添加延迟队列
	 * @param value 队列值
	 * @param delay 延迟时间
	 * @param queueCode 队列键
	 * @param <T> 泛型
	 */
	public <T> void addDelayQueue(T value, long delay, String queueCode) {
		addDelayQueue(value, delay, TimeUnit.SECONDS, queueCode);
	}

	/**
	 * 添加延迟队列
	 * @param value 队列值
	 * @param delay 延迟时间
	 * @param timeUnit 时间单位
	 * @param queueCode 队列键
	 * @param <T> 泛型
	 */
	public <T> void addDelayQueue(T value, long delay, TimeUnit timeUnit, String queueCode){
		try {
			RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
			RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
			delayedQueue.offer(value, delay, timeUnit);

			log.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
			//释放队列
			delayedQueue.destroy();
		} catch (Exception e) {
			log.error("(添加延时队列失败) {}", e.getMessage());
			throw new RuntimeException("(添加延时队列失败)");
		}
	}

	/**
	 * 获取延迟队列
	 * @param queueCode 队列主键
	 * @return value 队列值
	 */
	public <T> T getDelayQueue(@NonNull String queueCode) throws InterruptedException {
		RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
		//避免消息伪丢失（应用重启未消费），官网推荐
		redissonClient.getDelayedQueue(blockingDeque) ;
		return (T) blockingDeque.take();
	}

	/**
	 * 队列是否存在值
	 * @param value 队列值
	 * @param queueCode 队列主键
	 * @return boolean
	 */
	public boolean contains(Object value, String queueCode) {
		if (StringUtil.isBlank(queueCode) || Objects.isNull(value)) {
			return false;
		}
		try {
			RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
			RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
			boolean flag = delayedQueue.contains(value);
			//释放队列
			delayedQueue.destroy();
			return flag;
		} catch (Exception e) {
			log.error("(查询队列是否存在值异常) 队列键：{}，队列值：{},错误信息：{}",queueCode, value, e.getMessage());
			throw new RuntimeException("(查询队列是否存在值异常)");
		}
	}

	/**
	 * 删除延迟队列
	 * @param value 队列值
	 * @param queueCode 队列主键
	 * @return boolean
	 */
	public boolean removeDelayedQueue(Object value, String queueCode) {
		if (StringUtil.isBlank(queueCode) || Objects.isNull(value)) {
			return false;
		}
		try {
			RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
			RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
			boolean flag = delayedQueue.remove(value);
			if(flag){
				log.info("(删除延时队列保证唯一性) 队列键：{}，队列值：{}", queueCode, value);
			}
			//释放队列
			delayedQueue.destroy();
			return flag;
		} catch (Exception e) {
			log.error("(删除延时队列异常) 队列键：{}，队列值：{},错误信息：{}",queueCode, value, e.getMessage());
			throw new RuntimeException("(删除延时队列异常)");
		}
	}

}
